package com.im.tcp.receiver;

import com.alibaba.fastjson.JSONObject;
import com.ecode.imecode.pack.MessagePack;
import com.im.tcp.MQFactory;
import com.im.tcp.receiver.process.MsgProcessFactory;
import com.lld.im.common.constants.Constants;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.impl.AMQCommand;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 获得业务场通过mq发送到数据包
 */
@Slf4j
public class MQMsgReceiver {


    /**
     * 这里的brokerId在当前测试环境并没有使用 分布式场景下需要添加 用户区分服务
     */
    public static String brokerId = "";

    /**
     * 创建队列 绑定交换机 接收客户端消息
     */
    private static void startReceiverMsg() {
        /**
         * 这里的思路 在分布式情况下 根据自身的id创建 队列以及交换机 同时指定路由key
         * 当如果又另一个想要发送消息到这台服务器时 就可以给brokerId来指定交换机以及路由key
         * 目标服务器就可以监听队列 处理收到的消息
         */
        try {
            Channel channel = MQFactory.getChannel(Constants.RabbitConstants.MessageService2Im+brokerId);
            String queueName = Constants.RabbitConstants.MessageService2Im + brokerId;
            String exchangeName = Constants.RabbitConstants.MessageService2Im;
            String routeingKey = brokerId;
            //创建交换机
            channel.exchangeDeclare(exchangeName, "direct", true);
            //声明队列 队列名 是否持久化 是否排他 是否自动删除 其它参数
            channel.queueDeclare(queueName, true, false, false, null);
            //这里的队列和转换器name设置为同一个
            //绑定转换器 队列名 交换机 路由key (必须先创建/存在对于名字的交换机/队列才可以绑定 不然会出现io错误)
            //这里的路由key指定为brokerId 将消息转发到指定的服务器id上
            channel.queueBind(queueName, exchangeName, routeingKey);
            //创建消费者
            channel.basicConsume(queueName, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String s = new String(body);
                    log.info("接收到消息:{}", s);
                    MessagePack messagePack = JSONObject.parseObject(s, MessagePack.class);
                    //提交处理
                    boolean result
                            = MsgProcessFactory.getMsgProcess(AMQCommand.EMPTY_FRAME_SIZE).processMsg(messagePack);
                    if(result){
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            });
        } catch (IOException e) {
            log.error("接收消息失败!{}", e.getMessage());
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        }

    }


    public static void init(String brokerId) {
        if (!StringUtils.isBlank(brokerId)) {
            MQMsgReceiver.brokerId = brokerId;
        }
        startReceiverMsg();
    }

}
